package com.unitd.frame.comm.serializer.impl;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.unitd.frame.comm.serializer.ISerializer;

import java.io.IOException;
import java.util.Deque;
import java.util.concurrent.ConcurrentLinkedDeque;

/**
 * @desc 使用 KryoPool 实现序列化
 * @filename KryoPoolSerializer.java
 * @copyright www.unitd.com
 * @author Hudan
 * @version 1.0
 * @date 2016/11/28
 */
public class KryoPoolSerializer implements ISerializer {

	/**
	 * @desc Kryo 的包装
	 */
	private static class KryoHolder {
		private Kryo kryo;
		static final int BUFFER_SIZE = 1024;
		private Output output = new Output(BUFFER_SIZE, -1); 			//reuse
		private Input input = new Input();

		KryoHolder(Kryo kryo) {
			this.kryo = kryo;
		}
	}

	/**
	 * @desc kryo对象序列化池
	 * @filename KryoPool.java
	 * @copyright www.unitd.com
	 * @author Hudan
	 * @version 1.0
	 * @date 2017/2/18
	 */
	interface KryoPool {

		/**
		 * 获取一个kryo对象实例
		 * @return KryoHolder instance
		 */
		KryoHolder get();

		/**
		 * @desc 向队列末尾插入一个kryo对象
		 * @param kryo holder
		 */
		void offer(KryoHolder kryo);
	}

	/**
	 * @desc 由于kryo创建的代价相对较高, 这里使用空间换时间;对KryoHolder对象进行重用
	 * @filename KryoPoolImpl.java
	 * @copyright www.unitd.com
	 * @author Hudan
	 * @version 1.0
	 * @date 2017/2/18
	 */
	public static class KryoPoolImpl implements KryoPool {

		/**
		 * @desc Kryo对象池队列,默认是1500在线最大3K,默认是线程安全的队列
		 * default is 1500 online server limit 3K thread safe list
		 */
		private final Deque<KryoHolder> kryoHolderDeque = new ConcurrentLinkedDeque<>();

		private KryoPoolImpl() {

		}

		/**
		 * @desc 获取KryoPool对象实例
		 * @return KryoPool instance
		 */
		public static KryoPool getInstance() {
			return Singleton.pool;
		}

		/**
		 * @desc 创建一个新的kryo对象供应用使用
		 * @return KryoHolder instance
		 */
		public KryoHolder creatInstnce() {
			Kryo kryo = new Kryo();
			kryo.setReferences(false);
			return new KryoHolder(kryo);
		}

		/**
		 * 获取一个kryo对象实例
		 * @return KryoHolder instance
		 */
		@Override
		public KryoHolder get() {
			// 从这个表的头部获取和删除队列
			KryoHolder kryoHolder = kryoHolderDeque.pollFirst();
			return kryoHolder == null ? creatInstnce() : kryoHolder;
		}

		/**
		 * @desc 插入指定元素在这个队列的尾部,并将队列对象返回
		 * @param kryoHolder kryo 对象
		 */
		@Override
		public void offer(KryoHolder kryoHolder) {
			kryoHolderDeque.addLast(kryoHolder);
		}

		/**
		 * 创建一个单例KryoPool对象
		 */
		private static class Singleton {
			private static final KryoPool pool = new KryoPoolImpl();
		}
	}

	/**
	 * @desc 获取序列化类型名称
	 * @return
	 */
	@Override
	public String name() {
		return "kryo_pool_ser";
	}

	/**
	 * @desc 序列化指定对象
	 * @param obj 需要序列化的对象
	 * @return
	 * @throws IOException
	 */
	@Override
	public byte[] serialize(Object obj) throws IOException {
		KryoHolder kryoHolder = null;
		if (obj == null)
			throw new RuntimeException("obj can not be null");
		try {
			kryoHolder = KryoPoolImpl.getInstance().get();
			kryoHolder.output.clear(); 						// clear Output -->每次调用的时候 重置
			kryoHolder.kryo.writeClassAndObject(kryoHolder.output, obj);
			return kryoHolder.output.toBytes(); 			// 无法避免拷贝
		} catch (RuntimeException e) {
			throw new RuntimeException(e);
		} finally {
			if (kryoHolder != null) {
				KryoPoolImpl.getInstance().offer(kryoHolder);
			}
			// obj = null; //GC
		}
	}

	/**
	 * @desc 反序列化指定对象
	 * @param bytes 需要反序列化的对象
	 * @return
	 * @throws IOException
	 */
	@Override
	public Object deserialize(byte[] bytes) throws IOException {
		KryoHolder kryoHolder = null;
		if (bytes == null) throw new RuntimeException("bytes can not be null");
		try {
			kryoHolder = KryoPoolImpl.getInstance().get();
			kryoHolder.input.setBuffer(bytes, 0, bytes.length);
			return kryoHolder.kryo.readClassAndObject(kryoHolder.input);
		} catch (RuntimeException e) {
			throw new RuntimeException(e);
		} finally {
			if (kryoHolder != null) {
				KryoPoolImpl.getInstance().offer(kryoHolder);
			}
			// bytes = null; // for gc
		}
	}
}